package com.example.kafka.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Kafka 客户端管理类
 * 用于模拟连接异常和资源泄漏场景
 */
@Component
public class KafkaClientManager {
    private static final Logger logger = LoggerFactory.getLogger(KafkaClientManager.class);
    
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    // 当前活跃的 Kafka 客户端实例
    private volatile KafkaProducer<String, String> currentProducer;
    private final AtomicInteger clientCount = new AtomicInteger(0);
    
    /**
     * 获取当前 Kafka 客户端
     * 如果当前没有客户端或客户端连接异常，则创建新的客户端
     */
    public KafkaProducer<String, String> getProducer() {
        if (currentProducer == null) {
            synchronized (this) {
                if (currentProducer == null) {
                    currentProducer = createNewProducer();
                }
            }
        }
        return currentProducer;
    }
    
    /**
     * 创建新的 Kafka 客户端
     * 注意：这里故意不关闭旧的客户端，模拟资源泄漏
     */
    private KafkaProducer<String, String> createNewProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000);
        
        KafkaProducer<String, String> newProducer = new KafkaProducer<>(props);
        
        return newProducer;
    }
    
    /**
     * 在连接异常时创建新的客户端
     * 注意：这里故意不关闭旧的客户端，模拟资源泄漏
     */
    public void recreateProducer() {
        synchronized (this) {
            KafkaProducer<String, String> oldProducer = currentProducer;
            currentProducer = createNewProducer();
            logger.info("重新创建 Kafka 客户端，旧客户端未关闭");
        }
    }
} 